package com.moli.iov.gateway.server.listener;

import com.moli.iov.consumer.KafkaConsumerServer;
import com.moli.iov.gateway.server.NettyChannelManager;
import com.moli.iov.producer.KafkaProducerServer;
import com.moli.iov.protocol.model.dto.VehicleOnLineDto;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.Acknowledgment;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @author fudh
 * @ClassNmme SynVinsListener
 * @date 2019/4/11 17:52
 * @Description: 接收同步消息，将本节点的vin同步到redis中去
 */
public class SynVinsListener extends KafkaConsumerServer {

    private static final Logger logger = LoggerFactory.getLogger(SynVinsListener.class);
    @Autowired
    private KafkaProducerServer kafkaProducerServer;

    @Override
    public void manualHandle(ConsumerRecord<String, Object> consumerRecord, Acknowledgment acknowledgment) {

        NettyChannelManager nettyChannelManager = NettyChannelManager.getInstance();
        ConcurrentHashMap<String, String> activeVinChannelMap = nettyChannelManager.getActiveVinChannelMap();
        ConcurrentHashMap.KeySetView<String, String> keySetView = activeVinChannelMap.keySet();
        logger.error("开始同步，本节点在线数："+keySetView.size());
        VehicleOnLineDto vehicleOnLineDto = null;
        for (String vin : keySetView) {
            vehicleOnLineDto = new VehicleOnLineDto(vin, true, System.currentTimeMillis());
            kafkaProducerServer.sndMesForTemplate(NettyChannelManager.VEHICLE_ONLINE, vin, vehicleOnLineDto);
        }
        logger.error("同步完成！");
    }
}
